package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Stream_Join {
  def main(args: Array[String]): Unit = {
    //TODO SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    val dstream1: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9997 )
    val dstream2: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop103", 9998)
    //两个流的join，这个没啥滑头就直接将两个客户端的数据转换一下格式，然后join处理

    dstream1.map((_, 9997)).join(dstream2.map((_, 9998))).print()


    ssc.start()
    ssc.awaitTermination()

  }

}
